package cn.ac.iscas.service.transformer.statistics;

import cn.ac.iscas.base.utils.ParameterUtils;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;

import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/12/6 14:11
 */
public class StatisticsTransformer {
    private final static String ID = "statistics_trans";
    private final static String STATISTICS_KEY = "statistic";
    private final static String COLUMN_KEY = "column";
    private final static String OPERATOR_KEY = "operator";
    private final static String LABEL_KEY = "label";

    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<FlinkSourceColumn> frontColumns;

    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public StatisticsTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public StatisticsTransformer(String jobId, String taskId, String frontTableName, List<FlinkSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
    }


    public StatisticsTransformer() {

    }

    // SELECT MAX(age) as maxa, SUM(score) as suma, MIN(age) as mina, STDDEV_POP(score) as stdp, STDDEV_SAMP(score) as stds, VAR_POP(score) as varp, VAR_SAMP(age) as
    // vars FROM c.c.ods_classtable;
    public String getSql(Map<String, Object> parameters) {
        this.currentTableName = String.format("%s.%s.%s_%s", jobId, jobId, ID, UUID.randomUUID().toString().substring(0,6));
        String createViewSql = String.format("CREATE VIEW %s AS ", this.currentTableName);
        StringBuffer statisticsBuffer = new StringBuffer();

        List<Map<String, Object>> statisticsList = ParameterUtils.getRequiredList(parameters, STATISTICS_KEY);
        for (Map<String, Object> statistics : statisticsList) {
            String operator = (String)statistics.getOrDefault(OPERATOR_KEY, "");
            String column = (String)statistics.getOrDefault(COLUMN_KEY, "");
            String label = (String)statistics.getOrDefault(LABEL_KEY, "");
            if (column.isEmpty() || operator.isEmpty() || label.isEmpty()) {
                continue;
            }
            switch (operator.toUpperCase(Locale.ROOT)) {
                case "AVG":
                    statisticsBuffer.append(String.format("AVG(%s) AS %s , ", column, label));
                    break;
                case "MAX":
                    statisticsBuffer.append(String.format("MAX(%s) AS %s , ", column, label));
                    break;
                case "MIN":
                    statisticsBuffer.append(String.format("MIN(%s) AS %s , ", column, label));
                    break;
                case "SUM":
                    statisticsBuffer.append(String.format("SUM(%s) AS %s , ", column, label));
                    break;
                case "STDDEV_POP":
                    statisticsBuffer.append(String.format("STDDEV_POP(%s) AS %s , ", column, label));
                    break;
                case "STDDEV_SAMP":
                    statisticsBuffer.append(String.format("STDDEV_SAMP(%s) AS %s , ", column, label));
                    break;
                case "VAR_POP":
                    statisticsBuffer.append(String.format("VAR_POP(%s) AS %s , ", column, label));
                    break;
                case "VAR_SAMP":
                    statisticsBuffer.append(String.format("VAR_SAMP(%s) AS %s , ", column, label));
                    break;
                default:
                    break;
            }
        }

        String statisticsStr = statisticsBuffer.toString();
        statisticsStr = statisticsStr.trim().endsWith(",") ? statisticsStr.substring(0, statisticsStr.lastIndexOf(",")) : statisticsStr;
        String selectSql = String.format(" ( SELECT %s FROM %s ) ", statisticsStr, frontTableName);
        createViewSql += selectSql;
        updateSource();
        return createViewSql;
    }

    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
